/**
 * Copyright 2007-2008 Alex Tkachman
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.ait.actors;

import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;

public abstract class Continuation<V> extends AtomicInteger implements Callable<V> {
    private Continuation owner;
    private int ownerKey;
    private volatile CountDownLatch waiting;
    private boolean looping;

    protected Object [] results;

    public Continuation() {
    }

    private void setOwner (Continuation continuation, int ownerKey) {
        if (owner != null)
          throw new IllegalStateException();

        this.owner = continuation;
        this.ownerKey = ownerKey;
    }

    public V execute () throws Exception {
        Thread thread = Thread.currentThread();
        final boolean notWorkerThread = !(thread instanceof WorkerThread);
        try {
            if (notWorkerThread)
              waiting = new CountDownLatch(1);
            else
              looping = true;
            return doCall ();
        }
        catch (ControlException ce) {
            if (notWorkerThread)
              waiting.await();
            else {
                while (get() != 0)
                  ((WorkerThread)thread).workerStep();
            }
            return call ();
        }
        catch (Throwable t) {
            Scheduler.debug(t);
            throw new RuntimeException(t);
        }
    }

    protected final void schedule (Callable callable []) throws Exception {
        if (callable == null || callable.length == 0)
          return;

        results = new Object[callable.length];
        for (int i = callable.length-1; i >= 0; --i) {
            callable[i] = callable[i] instanceof Continuation ? (Continuation) callable[i] : new CallableAdapter(callable[i]);
            ((Continuation)callable[i]).setOwner(this, i);
            incrementAndGet();
        }
        Scheduler.schedule(callable);
        Scheduler.throwControl();
    }

    private void setResult (int key, Object value) {
        results [key] = value;
        if (decrementAndGet() == 0) {
            // last nested task completed
            // if we are sleeping we need wake up
            // otherwise schedule ourselves for execution
            if (waiting == null) {
                if (!looping)
                  Scheduler.schedule(this);
            }
            else {
                waiting.countDown();
            }
        }
    }

    protected boolean isNestedCompleted() {
        return results != null && get() == 0;
    }

    public final V call() throws Exception {
        V res = doCall ();
        if (owner != null) {
            owner.setResult(ownerKey, res);
            Scheduler.throwControl ();
        }
        else
          return res;

        throw new IllegalStateException();
    }

    protected abstract V doCall() throws Exception;

    static class CallableAdapter extends Continuation {
        private final Callable callable;

        public CallableAdapter(Callable callable) {
            this.callable = callable;
        }

        protected Object doCall() throws Exception {
            return callable.call();
        }

        public String toString() {
            return "Adapter("+callable+")";
        }
    }
}
